package cn.wangqi.personal.redis;

import redis.clients.jedis.Client;
import redis.clients.jedis.Connection;
import redis.clients.jedis.JedisPubSub;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.SafeEncoder;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;

import static redis.clients.jedis.Protocol.Keyword.*;
import static redis.clients.jedis.Protocol.Keyword.PSUBSCRIBE;
import static redis.clients.jedis.Protocol.Keyword.PUNSUBSCRIBE;

public abstract class PubSubListener extends JedisPubSub {

    @Override
    public void proceed(Client client, String... channels) {
        setClient(client);
        client.subscribe(channels);
        flushClient();
        handler(client);
    }
    public abstract <T extends Serializable> void onMessage(String channel,T msg);
    private void handler(Client client) {
        do {
            List<Object> reply = client.getRawObjectMultiBulkReply();
            final Object firstObj = reply.get(0);
            if (!(firstObj instanceof byte[])) {
                throw new JedisException("Unknown message type: " + firstObj);
            }
            final byte[] resp = (byte[]) firstObj;
            if (Arrays.equals(SUBSCRIBE.raw, resp)) {
                int subscribedChannels = ((Long) reply.get(2)).intValue();
                setSubscribedChannels(subscribedChannels);
                final byte[] bchannel = (byte[]) reply.get(1);
                final String strchannel = (bchannel == null) ? null
                        : SafeEncoder.encode(bchannel);
                onSubscribe(strchannel, subscribedChannels);
            } else if (Arrays.equals(UNSUBSCRIBE.raw, resp)) {
                int subscribedChannels = ((Long) reply.get(2)).intValue();
                setSubscribedChannels(subscribedChannels);
                final byte[] bchannel = (byte[]) reply.get(1);
                final String strchannel = (bchannel == null) ? null
                        : SafeEncoder.encode(bchannel);
                onUnsubscribe(strchannel, subscribedChannels);
            } else if (Arrays.equals(MESSAGE.raw, resp)) {
                final byte[] bchannel = (byte[]) reply.get(1);
                final byte[] bmesg = (byte[]) reply.get(2);
                final String strchannel = (bchannel == null) ? null
                        : SafeEncoder.encode(bchannel);
                final Serializable strmesg = (bmesg == null) ? null : RedisUtil
                        .decode(bmesg);
                onMessage(strchannel, strmesg);
            } else if (Arrays.equals(PMESSAGE.raw, resp)) {
                final byte[] bpattern = (byte[]) reply.get(1);
                final byte[] bchannel = (byte[]) reply.get(2);
                final byte[] bmesg = (byte[]) reply.get(3);
                final String strpattern = (bpattern == null) ? null
                        : SafeEncoder.encode(bpattern);
                final String strchannel = (bchannel == null) ? null
                        : SafeEncoder.encode(bchannel);
                final String strmesg = (bmesg == null) ? null : SafeEncoder
                        .encode(bmesg);
                onPMessage(strpattern, strchannel, strmesg);
            } else if (Arrays.equals(PSUBSCRIBE.raw, resp)) {
                int subscribedChannels = ((Long) reply.get(2)).intValue();
                setSubscribedChannels(subscribedChannels);
                final byte[] bpattern = (byte[]) reply.get(1);
                final String strpattern = (bpattern == null) ? null
                        : SafeEncoder.encode(bpattern);
                onPSubscribe(strpattern, subscribedChannels);
            } else if (Arrays.equals(PUNSUBSCRIBE.raw, resp)) {
                int subscribedChannels = ((Long) reply.get(2)).intValue();
                setSubscribedChannels(subscribedChannels);
                final byte[] bpattern = (byte[]) reply.get(1);
                final String strpattern = (bpattern == null) ? null
                        : SafeEncoder.encode(bpattern);
                onPUnsubscribe(strpattern, subscribedChannels);
            } else {
                throw new JedisException("Unknown message type: " + firstObj);
            }
        } while (isSubscribed());

	/* Invalidate instance since this thread is no longer listening */
        setClient(null);
	/*
	 * Reset pipeline count because subscribe() calls would have increased
	 * it but nothing decremented it.
	 */
         client.resetPipelinedCount();
    }
    private void setClient(Client client){
        try {
            Field clientField = JedisPubSub.class.getDeclaredField("client");
            clientField.setAccessible(true);
            clientField.set(this, client);
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
    }

    private void setSubscribedChannels(int value){
        try {
            Field subscribedChannelsField = JedisPubSub.class.getDeclaredField("subscribedChannels");
            subscribedChannelsField.setAccessible(true);
            subscribedChannelsField.set(this, value);
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
    }
    private void  flushClient(){
        try {
            Field clientField = JedisPubSub.class.getDeclaredField("client");
            clientField.setAccessible(true);
            Connection client = (Connection) clientField.get(this);
            Method method = Connection.class.getDeclaredMethod("flush");
            method.setAccessible(true);
            method.invoke(client);
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }
}
